package org.example.realtime.jtp.dws.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.example.realtime.jtp.common.utils.JdbcUtil;
import org.example.realtime.jtp.common.utils.KafkaUtil;

/**
 * @Title: JtpLogClickHouse
 * @Author Lianzy
 * @Package org.example.realtime.jtp.dws.log.job
 * @Date 2025/5/25 19:03
 * @description 将topic数据保存到clickhouse
 */
public class JtpTopiclogtocliHouse {
    public static void main(String[] args) throws Exception {
        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取页面浏览日志的tpic存入clickhouse
        DataStream<String> trafficdata = KafkaUtil.consumerKafka(env, "dwd-traffic-page-log");
//        savePageLog(trafficdata);
        //获取页面启动日志的tpic存入clickhouse
        DataStream<String> startdata = KafkaUtil.consumerKafka(env, "dwd-log-startT-log");
        savestartLog(startdata);
        //获取动作action日志的tpic存入clickhouse
        env.execute();
    }

    private static void savestartLog(DataStream<String> startdata) {
        /**
         * {
         *   "common": {
         *     "ar": "370000",
         *     "ba": "Honor",
         *     "ch": "wandoujia",
         *     "is_new": "1",
         *     "md": "Honor 20s",
         *     "mid": "eQF5boERMJFOujcp",
         *     "os": "Android 11.0",
         *     "uid": "76",
         *     "vc": "v2.1.134"
         *   },
         *   "start": {
         *     "entry": "icon",         --icon手机图标  notice 通知   install 安装后启动
         *     "loading_time": 18803,  --启动加载时间
         *     "open_ad_id": 7,        --广告页ID
         *     "open_ad_ms": 3449,    -- 广告总共播放时间
         *     "open_ad_skip_ms": 1989   --  用户跳过广告时点
         *   },
         *   "ts": 1585744304000
         * }
         */
        SingleOutputStreamOperator<String> map = startdata.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String s) throws Exception {
                        // 1解析JSON
                        JSONObject object = JSON.parseObject(s);
                        JSONObject commonObject = object.getJSONObject("common");
                        JSONObject startObject = object.getJSONObject("start");
                        // 2提前字段值
                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put("mid", commonObject.getString("mid"));
                        jsonObject.put("area_code", commonObject.getString("ar"));
                        jsonObject.put("brand", commonObject.getString("ba"));
                        jsonObject.put("channel", commonObject.getString("ch"));
                        jsonObject.put("is_new", commonObject.getString("is_new"));
                        jsonObject.put("model", commonObject.getString("md"));
                        jsonObject.put("os", commonObject.getString("os"));
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString());
                        jsonObject.put("version_code", commonObject.getString("vc"));

                        jsonObject.put("entry", startObject.getString("entry"));
                        jsonObject.put("loading_time", startObject.getLong("loading_time"));
                        jsonObject.put("open_ad_id", startObject.getLong("open_ad_id"));
                        jsonObject.put("open_ad_ms", startObject.getLong("open_ad_ms"));
                        jsonObject.put("open_ad_skip_ms", startObject.getLong("open_ad_skip_ms"));

                        jsonObject.put("ts", object.getLong("ts"));
                        // 3返回JSON字符串
                        return jsonObject.toJSONString();
                    }
                }
        );

        // 2-1字段名称
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "entry", "loading_time", "open_ad_id", "open_ad_ms", "open_ad_skip_ms",  "ts"
        };
// 2-2插入语句
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_start_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    entry, loading_time, open_ad_id, open_ad_ms, open_ad_skip_ms,\n" +
                "    ts\n" +
                ") VALUES (\n" +
                "     ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                " )" ;
// 2-3插入数据
        JdbcUtil.sinkToClickhouseUpsertJson(map,columns, upsertSql);


    }

    /**
     * 实时将页面浏览日志数据，写入ClickHouse表中
     */
    public static void savePageLog(DataStream<String> stream) {
    /*
       {
          "common": {
             "ar": "230000",
             "uid": "57",
             "os": "iOS 13.3.1",
             "ch": "Appstore",
             "is_new": "0",
             "md": "iPhone Xs",
             "mid": "mid_495168",
             "vc": "v2.1.134",
             "ba": "iPhone"
          },
          "page": {
             "page_id": "good_detail",
             "item": "4",
             "during_time": 19709,
             "item_type": "sku_id",
             "last_page_id": "good_list",
             "source_type": "query"
          },
          "ts": 1736784135000
       }
     */
        // 1数据转换
        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        // 1解析JSON
                        JSONObject object = JSON.parseObject(value);
                        JSONObject commonObject = object.getJSONObject("common");
                        JSONObject pageObject = object.getJSONObject("page");

                        // 2提前字段值
                        JSONObject jsonObject = new JSONObject() ;
                        jsonObject.put("mid", commonObject.getString("mid")) ;
                        jsonObject.put("area_code", commonObject.getString("ar")) ;
                        jsonObject.put("brand", commonObject.getString("ba")) ;
                        jsonObject.put("channel", commonObject.getString("ch")) ;
                        jsonObject.put("is_new", commonObject.getString("is_new")) ;
                        jsonObject.put("model", commonObject.getString("md")) ;
                        jsonObject.put("os", commonObject.getString("os")) ;
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString()) ;
                        jsonObject.put("version_code", commonObject.getString("vc")) ;
                        jsonObject.put("page_id", pageObject.getOrDefault("page_id", "")) ;
                        jsonObject.put("last_page_id", pageObject.getOrDefault("last_page_id", "")) ;
                        jsonObject.put("during_time", pageObject.getLong("during_time")) ;
                        jsonObject.put("sourceType", pageObject.getOrDefault("sourceType", "")) ;
                        jsonObject.put("item_type", pageObject.getOrDefault("item_type", "")) ;
                        jsonObject.put("item", pageObject.getOrDefault("item", "")) ;
                        jsonObject.put("ts", object.getLong("ts")) ;
                        // 3返回JSON字符串
                        return jsonObject.toJSONString();
                    }
                }
        );

        // 2-1字段名称
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "page_id", "last_page_id", "during_time", "sourceType", "item_type", "item", "ts"
        };
        // 2-2插入语句
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_page_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    page_id, last_page_id, during_time, sourceType, item_type, item,\n" +
                "    ts\n" +
                ") VALUES (\n" +
                "   ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                ")" ;
        // 2-3插入数据
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream,columns, upsertSql);
    }
}
